package com.itheima.flink.init;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author lilulu
 * @date 2023/2/27 11:16
 */

// 演示Flink的入门案例  -- 词频统计
public class FlinkWordCount {
    /**
     * 1) 创建Flink的流式计算核心环境类对象
     * 2) 添加Source数据源, 用于读取数据
     * 3) 添加相关的转换操作, 对数据进行分析处理
     * 4) 添加Sink组件, 将计算的结果进行输出操作
     * 5) 启动Flink程序
     */
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = executionEnvironment.socketTextStream("node1", 4444);

        // 后续端口来的数据, 可能是一行数据, 包含多个单词, 单词之间会用空格分开
        // 第一次监听到:  word hadoop hello word   --->  word,hadoop,hello,word
        //3.1: 将数据进行切割, 得到一个个单词, 并将单词附上 单词,1 操作
        SingleOutputStreamOperator<Tuple2<String,Long>> streamOperator = source.flatMap(new FlatMapFunction<String, Tuple2<String,Long>>() {

            public void flatMap(String line, Collector<Tuple2<String, Long>> collector) throws Exception {
                String[] words = line.split(" ");
                for (String word : words) {
                    collector.collect(new Tuple2<String, Long>(word,1L));
                }

            }
        });

        // 3.2: 根据 单词进行分组, 将每个组内的数值累加在一起即可
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = streamOperator.keyBy(0).sum(1);
        // 4) 添加Sink组件, 将计算的结果进行输出操作
        streamOperator.print();

        // 5) 启动Flink程序
        executionEnvironment.execute();
    }


}
